/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

package org.apache.airavata.metcat.db;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.ThriftKsDef;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.ColumnQuery;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SliceQuery;

import javax.swing.text.html.HTMLDocument;
import java.util.*;

public class ModelAPI {

    public void createKeyspace()
    {
        Cluster cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        KeyspaceDefinition newKeyspaceDef = HFactory.createKeyspaceDefinition(ModelConfig.KEYSPACE);
        if((cluster.describeKeyspace(ModelConfig.KEYSPACE)) == null){
            cluster.addKeyspace(newKeyspaceDef, true);
        }
    }

    public void createColumnFamily(String columnFamilyName) {
        Cluster cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        ColumnFamilyDefinition columnFamilyDefinition = HFactory.createColumnFamilyDefinition(ModelConfig.KEYSPACE, columnFamilyName, ComparatorType.BYTESTYPE);
        
        Boolean isAlreadyExists = false;
        KeyspaceDefinition keyspaceDef = cluster.describeKeyspace(ModelConfig.KEYSPACE);
        List<ColumnFamilyDefinition> columnFamilyList = keyspaceDef.getCfDefs();
        Iterator<ColumnFamilyDefinition> iterator = columnFamilyList.iterator();
	        while (iterator.hasNext()) {
             if(iterator.next().getName().toString().equals(columnFamilyDefinition.getName().toString())) {
                 isAlreadyExists = true;
                 break;
             }
	     }
        if(!isAlreadyExists){
        cluster.addColumnFamily(columnFamilyDefinition);
        }

    }

    // use to insert data to any column family
    public void InsertDataToColumnFamily(String columnFamily, String rowID, String columnName, String columnValue) {
        Cluster cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        Keyspace keyspace = HFactory.createKeyspace(ModelConfig.KEYSPACE, cluster);

        Mutator<String> mutator = HFactory.createMutator(keyspace, StringSerializer.get());

        mutator.addInsertion(rowID, columnFamily,
                HFactory.createStringColumn(columnName, columnValue));
        mutator.execute();
    }

    // Given the Row ID get the column value pairs of the entire row
    public HashMap<String,String> getAllRowData(String columnFamily, String rowID)
    {
        Cluster cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        Keyspace keyspace = HFactory.createKeyspace(ModelConfig.KEYSPACE, cluster);

        HashMap<String, String> resultMap = new HashMap<String, String>();

        SliceQuery query = HFactory.createSliceQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
            query.setColumnFamily(columnFamily)
            .setKey(rowID)
            .setRange("", "", false,Integer.MAX_VALUE);
        QueryResult<ColumnSlice<String, String>> result = query.execute();

        for (HColumn<String, String> column : result.get().getColumns()) {
            resultMap.put(column.getName(),column.getValue());
            //System.out.println(column.getName() +"::"+ column.getValue());
        }
        return resultMap;
    }
    //Given a row key and a column key range, retrieve list of data falling in that range
    public List<String> getAllDatafor(String columnFamily, String rowKey, String key1, String key2) {
        Cluster cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        Keyspace keyspace = HFactory.createKeyspace(ModelConfig.KEYSPACE, cluster);

        List<String> resultList = new ArrayList<String>();

        SliceQuery<String, String, String> query = HFactory.createSliceQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
        query.setColumnFamily(columnFamily).setKey(rowKey).setRange(key1, key2, false, 100);
        QueryResult<ColumnSlice<String, String>> result = query.execute();

        for (HColumn<String, String> column : result.get().getColumns()) {
            resultList.add(column.getValue());
            //System.out.println(column.getName() +"::"+ column.getValue());
        }
        return resultList;
    }


    public String getLastColumnValue(String columnFamily,String rowKey)
    {
        Cluster cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        Keyspace keyspace = HFactory.createKeyspace(ModelConfig.KEYSPACE, cluster);
        String lastColumnValue = null ;

        SliceQuery<String, String, String> query = HFactory.createSliceQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
        query.setColumnFamily(columnFamily).setKey(rowKey).setRange("", "", true, 1);
        QueryResult<ColumnSlice<String, String>> result = query.execute();
        for (HColumn<String, String> column : result.get().getColumns()) {
          lastColumnValue = column.getValue();
        }

        return lastColumnValue;


    }

}
